package ex.datastream.windows;

import ex.datastream.functions.sourceFunction.SourceFuncByEvent;
import ex.vo.Event;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Optional;

/**
 * ref:https://blog.csdn.net/wangpei1949/article/details/102996170
 */
public class CountEvictorTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.addSource(new SourceFuncByEvent())
                .keyBy(event -> 1)
                .windowAll(GlobalWindows.create())
                .trigger(
                        new Trigger<Event, GlobalWindow>() {
                            @Override
                            public TriggerResult onElement(Event element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {

                                ValueState<Boolean> firstSeen = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("first-seen", Types.BOOLEAN));
                                if (!Optional.ofNullable(firstSeen.value()).orElseGet(() -> false)) {
                                    long ts = ctx.getCurrentProcessingTime() + 5 * 1000L;
                                    SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                    System.out.println("【初始化定时器】 " + dateformat.format(ts));
                                    ctx.registerProcessingTimeTimer(ts);
                                    firstSeen.update(true);
                                }
                                System.out.printf("%s,",element.key);
                                return TriggerResult.CONTINUE;
                            }

                            @Override
                            public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
                                SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                System.out.println("\n【触发onProcessingTime】 = " +  dateformat.format(time));
                                ctx.registerProcessingTimeTimer(time +  5 * 1000L);
                                return TriggerResult.FIRE;
                            }

                            @Override
                            public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
                                return TriggerResult.CONTINUE;
                            }

                            @Override
                            public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {
                                ValueState<Boolean> firstSeen = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("first-seen", Types.BOOLEAN));
                                firstSeen.clear();
                            }
                        }
                )
                //每次只保留最新10条
                .evictor(CountEvictor.of(10))
                .process(
                        new ProcessAllWindowFunction<Event, String, GlobalWindow>() {
                            @Override
                            public void process(Context context, Iterable<Event> elements, Collector<String> out) throws Exception {
                                out.collect("窗口中有" + elements.spliterator().getExactSizeIfKnown() + "条元素");
                                System.out.println("****************************最新10条数据*****************************");
                                elements.forEach(a-> System.out.printf(a.key+","));
                                System.out.println("\n******************************************************************");
                            }
                        }
                )
                .print();
        env.execute();
    }
//【初始化定时器】 2024-02-29 16:25:14
//76,48,89,54,38,90,32,56,16,69,95,31,12,57,46,86,58,48,61,60,10,83,86,5,73,
//【触发onProcessingTime】 = 2024-02-29 16:25:14
//窗口中有10条元素
//****************************最新10条数据*****************************
//86,58,48,61,60,10,83,86,5,73,
//******************************************************************
//1,46,49,78,77,81,65,92,3,69,74,94,72,16,30,60,82,19,60,71,32,19,45,
//【触发onProcessingTime】 = 2024-02-29 16:25:19
//窗口中有10条元素
//****************************最新10条数据*****************************
//16,30,60,82,19,60,71,32,19,45,
}

